use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use anyhow::Result;

use protoactor_rs::actor::{Actor, ActorContext, ActorError, DefaultActorContext, PID, Props};
use protoactor_rs::system::ActorSystem;
use protoactor_rs::remote::{Remote, RemoteConfig, Config as RemoteConfigBuilder};
use protoactor_rs::cluster::{
    Grain, GrainContext, GrainState, Member, MemberStatus,
    ClusterSystem, ClusterConfig, ClusterEvent, GrainManager,
    ConsistentHashRing,
};

// 定义一个简单的虚拟Actor
struct HelloGrain {
    id: String,
    counter: usize,
}

// 虚拟Actor的消息
struct Hello {
    name: String,
}

// 虚拟Actor的响应
#[derive(Clone, Debug)]
struct HelloResponse {
    message: String,
    count: usize,
    node_id: String,
}

impl HelloGrain {
    fn new(id: &str) -> Self {
        Self {
            id: id.to_string(),
            counter: 0,
        }
    }
}

#[async_trait]
impl Actor for HelloGrain {
    type Context = GrainContext;
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn std::any::Any + Send>) -> Result<(), ActorError> {
        // 更新活动状态
        ctx.update_activity();
        
        if let Some(hello) = msg.downcast_ref::<Hello>() {
            let mut grain = self.clone();
            grain.counter += 1;
            
            // 获取当前节点ID（示例中简化处理）
            let node_id = std::env::var("NODE_ID").unwrap_or_else(|_| "unknown".to_string());
            
            let response = HelloResponse {
                message: format!("Hello, {}! (from grain {})", hello.name, self.id),
                count: grain.counter,
                node_id,
            };
            
            ctx.respond(response);
        } else if let Some(cmd) = msg.downcast_ref::<&str>() {
            if *cmd == "get_state" {
                ctx.respond(GrainState::Active);
            }
        }
        
        Ok(())
    }
}

#[async_trait]
impl Grain for HelloGrain {
    fn id(&self) -> &str {
        &self.id
    }
    
    fn kind(&self) -> &str {
        "hello_grain"
    }
    
    async fn activate(&mut self) -> Result<()> {
        println!("Activating grain: {}", self.id);
        Ok(())
    }
    
    async fn deactivate(&mut self) -> Result<()> {
        println!("Deactivating grain: {}", self.id);
        Ok(())
    }
}

impl Clone for HelloGrain {
    fn clone(&self) -> Self {
        Self {
            id: self.id.clone(),
            counter: self.counter,
        }
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // 初始化日志
    tracing_subscriber::fmt::init();
    
    // 模拟两个节点的集群
    if std::env::args().nth(1) == Some("node1".to_string()) {
        run_node("node1", 8090).await?;
    } else if std::env::args().nth(1) == Some("node2".to_string()) {
        run_node("node2", 8091).await?;
    } else {
        println!("Usage: cargo run --example cluster_hash -- [node1|node2]");
        println!("       Run two instances with different node IDs to simulate a cluster");
    }
    
    Ok(())
}

async fn run_node(node_id: &str, port: u16) -> Result<()> {
    // 设置节点ID环境变量
    std::env::set_var("NODE_ID", node_id);
    
    println!("Starting node: {}", node_id);
    
    // 创建actor系统
    let system = ActorSystem::new();
    
    // 创建远程配置
    let remote_config = RemoteConfigBuilder::new()
        .with_host(&format!("127.0.0.1:{}", port))
        .build();
    
    // 创建远程处理模块
    let mut remote = Remote::new(system.clone(), remote_config);
    remote.start().await?;
    
    // 注册HelloGrain
    remote.register::<HelloGrain>("hello_grain", Props::new_fn(|| Box::new(HelloGrain::new("temp"))))?;
    
    // 创建集群配置
    let cluster_config = ClusterConfig {
        name: "hello_grain".to_string(),  // 设置角色为hello_grain
        host: "127.0.0.1".to_string(),
        port,
        heartbeat_interval: 1,
        unreachable_timeout: 5,
        down_timeout: 10,
    };
    
    // 创建集群系统
    let (mut cluster, mut cluster_events) = ClusterSystem::new_with_memory_provider(
        cluster_config, 
        Arc::new(remote)
    );
    
    // 启动集群系统
    cluster.start().await?;
    
    // 创建Grain管理器
    let grain_manager = Arc::new(GrainManager::new(
        Arc::new(cluster.clone()),
        Duration::from_secs(30),  // 空闲超时
        Duration::from_secs(5),   // 清理间隔
    ));
    
    // 启动Grain管理器
    grain_manager.start().await;
    
    // 监听集群事件
    tokio::spawn(async move {
        while let Some(event) = cluster_events.recv().await {
            println!("Cluster event: {:?}", event);
        }
    });
    
    println!("Node {} started, press Ctrl+C to exit", node_id);
    
    // 如果是节点1，启动一个客户端任务发送请求
    if node_id == "node1" {
        // 等待节点2启动
        tokio::time::sleep(Duration::from_secs(2)).await;
        
        for i in 1..=10 {
            let grain_id = format!("grain{}", i % 5); // 使用5个不同的grain
            
            match grain_manager.get_grain::<HelloGrain>(
                "hello_grain",
                &grain_id,
                Props::new_fn(move || Box::new(HelloGrain::new(&grain_id))),
            ).await {
                Ok(pid) => {
                    // 发送10个请求给每个grain
                    for j in 1..=2 {
                        let msg = Hello { name: format!("User {}-{}", i, j) };
                        
                        match pid.request::<HelloResponse>(msg, Duration::from_secs(5)).await {
                            Ok(response) => {
                                println!("Grain {}: Response from node {}: {:?}", 
                                    grain_id, response.node_id, response);
                            },
                            Err(e) => {
                                println!("Error: {}", e);
                            }
                        }
                        
                        // 稍微等待
                        tokio::time::sleep(Duration::from_millis(200)).await;
                    }
                }
                Err(e) => {
                    println!("Error getting grain: {}", e);
                }
            }
        }
        
        // 让程序继续运行以便查看结果
        tokio::time::sleep(Duration::from_secs(10)).await;
    } else {
        // 对于节点2，永久运行并处理请求
        loop {
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    }
    
    Ok(())
} 